package com.pie4cloud.pie.bus.config;

import com.alibaba.nacos.api.naming.NamingFactory;
import com.alibaba.nacos.api.naming.NamingService;
import com.alibaba.nacos.api.naming.listener.Event;
import com.alibaba.nacos.api.naming.listener.EventListener;
import com.alibaba.nacos.api.naming.listener.NamingEvent;
import com.alibaba.nacos.api.naming.pojo.Instance;
import com.pie4cloud.pie.bus.service.KafkaManageService;
import com.pie4cloud.pie.common.core.util.RedisLock;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

@Service
public class ServiceStatusListner {
	@Value("${spring.cloud.nacos.discovery.server-addr}")
	private String serverAddr;
	private int instanceNum=0;
	@Autowired
	private KafkaManageService kafkaManageService;
	@Autowired
	private RedisLock redisLock;

	public void init() throws Exception {

		Properties properties = System.getProperties();
		properties.setProperty("serverAddr", serverAddr);
		properties.setProperty("namespace", "public");
		NamingService naming = NamingFactory.createNamingService(properties);
		List<String> serviceNames = new ArrayList<>();
		serviceNames.add("pie-bus-biz");
		for(String serviceName : serviceNames) {
			naming.subscribe(serviceName, new EventListener() {
				@Override
				public void onEvent(Event event) {
					boolean flag=redisLock.tryLock("pie-bus-biz");
					if(flag){
						try {
							List<Instance> instances =  ((NamingEvent)event).getInstances();
							//String serviceName = ((NamingEvent)event).getServiceName();
							if(instances.size()>=instanceNum){
								instanceNum=instances.size();
							}else{
								kafkaManageService.resetConsumer();
							}
						} catch (Exception e) {
							e.printStackTrace();
						} finally {
							redisLock.delete("pie-bus-biz");
						}
					}

				}
			});
		}
	}
}
